package tool;

import io.reactivex.*;
import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import joinery.DataFrame;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.poi.hssf.record.aggregates.CFRecordsAggregate;

import java.io.IOException;
import java.util.*;

import static tool.TableProperty.*;

public class DataProvider {
    public static Map<String,DataFrame<String>> cache;

    public static String[] parse(String...o)
    {
        return o;
    }

    public static Observable<ResultScanner> catchAllFromType(String type)
    {
//        SingleColumnValueFilter scvf = new SingleColumnValueFilter(
//                Bytes.toBytes(COLFAM),
//                Bytes.toBytes(TYPE),
//                CompareFilter.CompareOp.EQUAL,
//                Bytes.toBytes(type));
//        scvf.setFilterIfMissing(true);//过滤器Filter
        Observable<ResultScanner> resultScannerObservable = null;
        try {
            return HBaseTool.GetInstance().scanData(//获取数据，返回ResultScanner
                    TABLE_NAME,
                    DataProvider.parse(),//筛选需要的属性列
                    getStartRow(type),
                    getEndRow(type),
                    null);

        } catch (IOException e) {
            e.printStackTrace();
        }
        return resultScannerObservable;
    }
    public static Observable<ResultScanner> catchAllFromType(String type,String sid)
    {
//        SingleColumnValueFilter scvf = new SingleColumnValueFilter(
//                Bytes.toBytes(COLFAM),
//                Bytes.toBytes(TYPE),
//                CompareFilter.CompareOp.EQUAL,
//                Bytes.toBytes(type));
//        scvf.setFilterIfMissing(true);//过滤器Filter
        Observable<ResultScanner> resultScannerObservable = null;
        try {
            return HBaseTool.GetInstance().scanData(//获取数据，返回ResultScanner
                    TABLE_NAME,
                    DataProvider.parse(),//筛选需要的属性列
                    sid,
                    sid,
                    null);

        } catch (IOException e) {
            e.printStackTrace();
        }
        return resultScannerObservable;
    }
    private static String getStartRow(String type)
    {
        if (type.equals(S))
        {
            return "s200900000000";
        }else if (type.equals(T))
        {
            return "t100000";
        }else if (type.equals(C))
        {
            return "c300000";
        }else if (type.equals(SC))
        {
            return "sc200900000000";
        }else if (type.equals(TC))
        {
            return "tc300000";
        }
        return null;
    }
    private static String getEndRow(String type)
    {
        if (type.equals(S))
        {
            return "s209999999999";
        }else if (type.equals(T))
        {
            return "t109999";
        }else if (type.equals(C))
        {
            return "c309999";
        }else if (type.equals(SC))
        {
            return "sc209999999999";
        }else if (type.equals(TC))
        {
            return "tc309999";
        }
        return null;
    }
    private static void mapData(final List<String> head, final DataFrame<String> result, Observable<ResultScanner> toMap)
    {
        mapData(head,result,toMap,true);
    }
    private static void mapRData(final String type, final List<String> head, final DataFrame<String> result, Observable<ResultScanner> toMap, final boolean addHead)
    {
        toMap.concatMap(new Function<ResultScanner, ObservableSource<Result>>() {
            @Override
            public ObservableSource<Result> apply(ResultScanner results) throws Exception {
                return Observable.fromIterable(results);
            }
        }).map(new Function<Result, List<List<String>>>() {
            @Override
            public List<List<String>> apply(Result result) throws Exception {
                Cell[] cells = result.rawCells();
                List<List<String>> datas = new LinkedList<>();

                String id = CellTool.GetValue(result,COLFAM,ID);
                String rowKey = CellTool.GetRowName(result);
                if (type.equals(SC))
                {
                    for (Cell c:cells)
                    {
                        String key = CellTool.GetColName(c);
                        String value = CellTool.GetValue(c);

                        if (!key.equals(ID) && !key.equals(TYPE))
                        {
                            List<String> line = emptyList(6);
                            line.set(0,rowKey);
                            line.set(1,id);
                            line.set(5,type);

                            String[] cid_tid_time = key.split("_");
//                            Lg.info(cid_tid_time);
                            if(cid_tid_time.length<2)
                            {
                                continue;
                            }
                            line.set(2,cid_tid_time[0]);
                            line.set(3,value);
                            line.set(4,cid_tid_time[1]);
                            datas.add(line);
                        }
                    }
                }else if (type.equals(TC))
                {
                    for (Cell c:cells)
                    {
                        String key = CellTool.GetColName(c);
//                        String value = CellTool.GetValue(c);
                        if (!key.equals(ID) && !key.equals(TYPE))
                        {
                            List<String> line = emptyList(4);
                            line.set(0,rowKey);
                            line.set(1,id);
                            line.set(2,key);
                            line.set(3,type);
                            datas.add(line);
                        }
                    }
                }

                return datas;
            }
        }).concatMap(new Function<List<List<String>>, ObservableSource<List<String>>>() {
            @Override
            public ObservableSource<List<String>> apply(List<List<String>> lists) throws Exception {
                return Observable.fromIterable(lists);
            }
        }).subscribe(new Consumer<List<String>>() {
            @Override
            public void accept(List<String> strings) throws Exception {
//                Lg.info(strings);
                result.append(strings);
            }
        });

    }
    private static List<String> emptyList(int len)
    {
        List<String> result = new LinkedList<>();
        for (int i = 0;i<len;i++)
        {
            result.add("");
        }
        return result;
    }
    private static void mapData(final List<String> head, final DataFrame<String> result, Observable<ResultScanner> toMap, final boolean addHead)
    {
        toMap.concatMap(new Function<ResultScanner, ObservableSource<Result>>() {
            @Override
            public ObservableSource<Result> apply(ResultScanner results) throws Exception {
                return Observable.fromIterable(results);
            }
        }).map(new Function<Result, List<String>>() {
            @Override
            public List<String> apply(Result result) throws Exception {
                List<String> alignData = DataProvider.getAlignData(head, result,addHead);
                return alignData;
            }
        }).subscribe(new Consumer<List<String>>() {
            @Override
            public void accept(List<String> strings) throws Exception {
                result.append(strings);
            }
        });
    }


    public static DataFrame<String> loadAllFromType(String type)
    {
//        Lg.info("loadAllFromType(String type)");
        String[] typeLabel = TableProperty.TYPE_MAP.get(type);
        final List<String> cHeader = Arrays.asList(typeLabel);
        final DataFrame<String> result = new DataFrame<>();
        result.add(typeLabel);
        Observable<ResultScanner> cResult = catchAllFromType(type);
        if (type.equals(SC) || type.equals(TC))
        {
            mapRData(type,cHeader,result,cResult,false);
        }else
        {
            mapData(cHeader,result,cResult,false);
        }

        return result;
    }

    public static DataFrame<String> provideClasses()
    {
        final List<String> cHeader = Arrays.asList(ID, NAME, FCID, CREDIT);
        final DataFrame<String> classes = new DataFrame<>();
        classes.add(ID, CID, NAME, FCID, CREDIT);

        Observable<ResultScanner> cResult = catchAllFromType(C);
        mapData(cHeader,classes,cResult);
        return classes;
    }
    public static DataFrame<String> provideSC()
    {
        final List<String> cHeader = Arrays.asList(ID, CID);
        DataFrame<String> classes = new DataFrame<>();
        classes.add(ID,SID,CID,SCORE,TID,TYPE);

        Observable<ResultScanner> cResult = catchAllFromType(SC);
        mapRData(SC,cHeader,classes,cResult,false);
        classes = classes.drop(TYPE,SCORE);
        return classes;
    }

    public static DataFrame<String> provideSC(String sid)
    {
        sid = "sc"+sid.replaceAll("^[a-zA-Z]*","");

        final List<String> cHeader = Arrays.asList(ID, CID);
        DataFrame<String> classes = new DataFrame<>();
        classes.add(ID,SID,CID,SCORE,TID,TYPE);

        Observable<ResultScanner> cResult = catchAllFromType(SC,sid);
        mapRData(SC,cHeader,classes,cResult,false);
        classes = classes.drop(TYPE,SCORE);
        return classes;
    }
    public static DataFrame<String> provideTC()
    {
        final List<String> cHeader = Arrays.asList(ID, TID);
        final DataFrame<String> classes = new DataFrame<>();
        classes.add(ID,CID,TID);

        Observable<ResultScanner> cResult = catchAllFromType(TC);
        mapRData(TC,cHeader,classes,cResult,false);
        return classes;
    }
    public static DataFrame<String> provideTeacher()
    {
        final List<String> tHeader = Arrays.asList(ID, NAME);
        final DataFrame<String> teacher = new DataFrame<>();
        teacher.add(ID, TID, NAME);

        Observable<ResultScanner> cResult = catchAllFromType(T);
        mapData(tHeader,teacher,cResult);

        return teacher;
    }


    public static DataFrame<String> provideClass()
    {
        DataFrame<String> tc = provideTC();
        DataFrame<String> teacher = provideTeacher();
        DataFrame<String> classes = provideClasses();
        tc = tc.drop(ID);
        teacher = teacher.drop(ID);

        System.out.println(teacher);
        System.out.println(tc);

        final Map<String,String> tid_tnameMap = new HashMap<>();
        Observable.fromIterable(teacher.map().values())
                .subscribe(new Consumer<List<String>>() {
                    @Override
                    public void accept(List<String> strings) throws Exception {
//                        Lg.info(strings);
                        tid_tnameMap.put(strings.get(0),strings.get(1));
                    }
                });
        final Map<String,String> cid_tidMap = new HashMap<>();
        Observable.fromIterable(tc.map().values())
                .subscribe(new Consumer<List<String>>() {
                    @Override
                    public void accept(List<String> strings) throws Exception {
//                        Lg.info(strings);
                        cid_tidMap.put(strings.get(0),strings.get(1));
                    }
                });

        classes.add(TID,TNAME);
        ArrayList<Object> head = new ArrayList<>(classes.columns());
        final int tidIndex = head.indexOf(TID);
        final int tnameIndex = head.indexOf(TNAME);
        int cidIndex = head.indexOf(CID);


        Set<Object> index = classes.index();
        final DataFrame<String> finalClasses = classes;
        Observable.fromIterable(index)
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {
                        String cid = finalClasses.get(o, CID);
                        finalClasses.set(o,TID,cid_tidMap.get(cid));
                        String tid = finalClasses.get(o, TID);
                        finalClasses.set(o,TNAME,tid_tnameMap.get(tid));
                    }
                });

        return finalClasses;
    }

    public static DataFrame<String> provideStudentId()
    {
        final List<String> tHeader = Arrays.asList(ID, NAME);
        final DataFrame<String> teacher = new DataFrame<>();
        teacher.add(ID, SID, NAME);

        Observable<ResultScanner> cResult = catchAllFromType(S);
        mapData(tHeader,teacher,cResult);

        return teacher;
    }


    public static List<String> getAlignData(List<String> head,Result result,boolean addHeadInFirst)
    {
        List<String> data = new ArrayList<>();
        Cell[] cells = result.rawCells();
        for (Cell cell:cells)
        {
            String colName = CellTool.GetColName(cell);
            String value = CellTool.GetValue(cell);
            int index = head.indexOf(colName);
            if (index != -1)
            {
                while (index+1>data.size())
                {
                    data.add("");
                }
                data.set(index,value);
            }
        }
        String rowName = CellTool.GetRowName(result);
        if (addHeadInFirst)
        {
            data.add(0,rowName);

        }else
        {
            data.set(0,rowName);
        }

        return data;
    }

    public static List<String> getAlignData(List<String> head,Result result)
    {
        return getAlignData(head,result,false);
    }
}
